package br.pucrs.lognplay.hadoop;

import org.apache.log4j.Logger;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.ImmutableSettings.Builder;
import org.elasticsearch.common.transport.InetSocketTransportAddress;

public class ElasticSearchClient {

	private Client esClient;

	private Builder settings;

	private String indexName;

	private String type;

	private BulkRequestBuilder bulkRequest;

	private int bufferSize = 0;

	private int ttl = 0;

	// private static final String ES_HOST = "10.252.67.6";
	private static final String ES_HOST = "ec2-177-71-224-71.sa-east-1.compute.amazonaws.com";

	private static final Logger logger = Logger.getLogger(ElasticSearchClient.class);

	public ElasticSearchClient() {

	}

	public void startClient() {
		logger.trace("Starting ElasticSearchClient");

		settings = ImmutableSettings.settingsBuilder();
		settings.put("cluster.name", "lognplayESCluster");
		settings.put("node.data", false);
		settings.build();

		setEsClient(new TransportClient(settings).addTransportAddress(new InetSocketTransportAddress(ES_HOST, 9300)));
	}

	public void stopClient() {
		logger.trace("Stopping ElasticSearchClient");

		if (isBulkRequest() && hasOpenBulkRequest()) {
			logger.trace("Has one opened bulk.");
			closeBulkRequest();
		}

		getEsClient().close();
	}

	public void put(String json) {

		try {
			if (isBulkRequest()) {

				IndexRequestBuilder indexBuilder = getEsClient().prepareIndex(getIndexName(), getType()).setSource(json);
				if (getTtl() > 0) {
					indexBuilder.setTTL(getTtl());
				}
				getBulkRequest().add(indexBuilder);

				/**
				 * If actions > buffer size then create a new bulkrequest
				 */
				if (getBulkRequest().numberOfActions() >= bufferSize) {
					logger.trace("Buffer overflows! Creating a new bulk");
					closeBulkRequest();
				}
			} else {
				IndexResponse response = getEsClient().prepareIndex(indexName, type).setSource(json).execute().actionGet();
			}
		} catch (Exception e) {
			logger.error(e);
		}
	}

	private BulkRequestBuilder getBulkRequest() {
		if (bulkRequest == null) {
			logger.trace("Opening a new Bulk. (BufferSize: " + getBufferSize() + ")");
			bulkRequest = getEsClient().prepareBulk();
		}
		return bulkRequest;
	}

	private void resetBulkRequest() {
		bulkRequest = null;
	}

	private void closeBulkRequest() {
		try {
			logger.trace("Closing Bulk.");
			BulkResponse bulkResponse = bulkRequest.execute().actionGet();
			resetBulkRequest();
		} catch (Exception e) {
			logger.error(e);
		}
	}

	private boolean isBulkRequest() {
		return (getBufferSize() > 0);
	}

	private boolean hasOpenBulkRequest() {
		return (bulkRequest != null);
	}

	public String getIndexName() {
		return indexName;
	}

	public void setIndexName(String indexName) {
		this.indexName = indexName;
	}

	public String getType() {
		return type;
	}

	public void setType(String type) {
		this.type = type;
	}

	public int getBufferSize() {
		return bufferSize;
	}

	public void setBufferSize(int bufferSize) {
		this.bufferSize = bufferSize;
	}

	public int getTtl() {
		return ttl;
	}

	public void setTtl(int ttl) {
		this.ttl = ttl;
	}

	public Client getEsClient() {
		if (esClient == null) {
			startClient();
		}
		return esClient;
	}

	public void setEsClient(Client client) {
		this.esClient = client;
	}

}
